package zhongche.config;

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Properties;

public class BackgroundConfig {
     private static Logger LOGGER = LoggerFactory.getLogger(BackgroundConfig.class);

    private static final String PROPERTIES_FILE = "backconfig.properties";

    private static Properties POP = new Properties();

    public static final String KAFKA_URI;

    public static final String KAFKA_TOPIC_GUANGZHOU_2;

    static {
        loadProperties();
        KAFKA_URI = POP.getProperty("kafka.url", null);
        KAFKA_TOPIC_GUANGZHOU_2 = POP.getProperty("kafka.topic.guangzhou2");


    }


    /**
     * 读取配置文件
     */
    private static void loadProperties(){
        try {
            POP.load(BackgroundConfig.class.getClassLoader().getResourceAsStream(PROPERTIES_FILE));
        } catch (IOException e) {
            LOGGER.info("读取配置文件"+PROPERTIES_FILE+"失败！！");
            e.printStackTrace();
        }
    }

    /**
     * 获取kafka消费者配置
     * @param kafkaUri
     * @param groupId
     * @return
     */
    public static Properties getKafkaConsumerProperties(String kafkaUri,String groupId) {
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", kafkaUri);
        properties.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("group.id",groupId);
        return properties;
    }

//    public static FlinkKafkaConsumer<String> getFlinkKafkaConsumer(String kafkaUri,String groupId, String topic){
//
//    }

}
